package main

import (
	"fmt"
	"net"
	"os"
	"time"
	"strconv"
	"bytes"
	"encoding/binary"
	"sync"
	"sync/atomic"
)

var g_counter int64 = 0
var g_connect int64 = 0

var g_mu   sync.Mutex
var g_cond *sync.Cond

func main() {

	var host string = "127.0.0.1"
	var port int = 6666
	var max_connects int = 1

	var err error

	g_mu   = sync.Mutex{}
	g_cond = sync.NewCond(&g_mu)

	if len(os.Args) >= 2 {
		max_connects, err = strconv.Atoi(os.Args[1])
		if err != nil {
			fmt.Printf("convert param:connects fail, value:%s\n", os.Args[1])
			return
		}
	}
	if len(os.Args) >= 3 {
		host = os.Args[2]
	}
	if len(os.Args) >= 4 {
		port, err = strconv.Atoi(os.Args[3])
		if err != nil {
			fmt.Printf("convert param:port fail, value:%s\n", os.Args[3])
			return
		}
	}

	fmt.Printf("param.max_connects:%d\n", max_connects)
	fmt.Printf("param.host:%s\n", host)
	fmt.Printf("param.port:%d\n", port)

	go func() {
		fmt.Printf("[create socket coroutine] running ...\n")
		var id int = 1
		for {
			count := int64(max_connects) - atomic.LoadInt64(&g_connect)
			fmt.Printf("need to create %d coroutine\n", count)
			for i := 0; i < int(count); i++ {
				atomic.AddInt64(&g_connect, 1)
				go echo_runner(host, port, id)
				id = id + 1
			}

			g_cond.L.Lock()
			fmt.Printf("[create socket coroutine] pass lock\n")
			for atomic.LoadInt64(&g_connect) >= int64(max_connects) {
				g_cond.Wait()
				fmt.Printf("[create socket coroutine] wait pass, cur_connect:%v, max_connect:%v\n", g_connect, max_connects);
			}
			g_cond.L.Unlock()
			fmt.Printf("[create socket coroutine] pass unlock\n")
		}
	} ()

	var last_counter int64 = 0
	for {
		time.Sleep(time.Duration(1) * time.Second)
		var now_counter int64 = atomic.LoadInt64(&g_counter)
		fmt.Printf("[%s] qps:%d\n", time.Now().Format("2006-01-02 15:04:05"), now_counter - last_counter) 
		last_counter = now_counter
	}
}

func echo_runner(host string, port int, id int) {

	addr := fmt.Sprintf("%v:%v", host, port)
	conn, err := net.Dial("tcp", addr)
	if  err != nil {
		fmt.Printf("client connect fail, id:%v, addr:%v, error:%v\n", id, addr, err)
		return
	}
	fmt.Printf("client connect succeed, id:%v\n", id)

	c, ok := conn.(*net.TCPConn)
	if !ok {
		panic("socket invalid");
	}
	c.SetNoDelay(true)

	send_msg := fmt.Sprintf("user_%v|hello_world", id)	

	err = send_packet(conn, send_msg)
	if err != nil {
		fmt.Printf("client send content fail, id:%d, content:%s, error:%v\n", id, send_msg, err)
		return
	}
	fmt.Printf("client send_packet succeed, id:%v, content:%v, size:%v\n", id, send_msg, len(send_msg))

	var recv_msg string
	err = recv_packet(conn, &recv_msg)
	if err != nil {
		fmt.Printf("client recv content fail, id:%d, error:%v\n", id, err)
		return
	}

	if send_msg != recv_msg {
		fmt.Printf("client error, send_msg != recv_msg, id:%d", id)
		os.Exit(-1)
	}

	conn.Close()

	atomic.AddInt64(&g_counter, 1)
	atomic.AddInt64(&g_connect, -1)

	g_cond.Signal()
}

func send_packet(conn net.Conn, content string) error {

	var size int32 = int32(len(content))

	buf := bytes.NewBuffer([]byte{})

	binary.Write(buf, binary.BigEndian, size)
	binary.Write(buf, binary.BigEndian, []byte(content))

	return send_full(conn, buf.Bytes())
}

func recv_packet(conn net.Conn, content *string) error {

	header := make([]byte, 4)
	err := recv_full(conn, header)
	if err != nil {
		return err
	}

	size := int32(binary.BigEndian.Uint32(header))

	body := make([]byte, size)
	err = recv_full(conn, body)
	if err != nil {
		return err
	}

	*content = string(body[:])

	return nil
}

func send_full(conn net.Conn, buf []byte) error {

	pos  := 0 
	size := len(buf)

	for {
		if pos >= size {
			break
		}

		ret, err := conn.Write(buf[pos:])
		if  err != nil {
			return err
		}

		pos += ret
	}
	return nil
}

func recv_full(conn net.Conn, buf []byte) error {

	pos  := 0
	size := len(buf)

	for {
		if pos >= size {
			break
		}

		ret, err := conn.Read(buf[pos:])
		if  err != nil {
			return err
		}

		pos += ret
	}
	return nil
}

